package cn.tecnova.analysis

import java.sql.{Connection, DriverManager, PreparedStatement, ResultSet}
import java.util.UUID

import cn.tecnova.bean.{BaAnalysisBean, BaArticle, BaUserWarning, NlpJsonBean}
import cn.tecnova.utils.{BroadcastKafkaProducer, BroadcastTools, ConfigHandler, DiyRelationArticleUtils}
import com.alibaba.fastjson.JSON
import com.google.gson.Gson
import org.apache.commons.lang.StringUtils
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.log4j.{Level, Logger}
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.dstream.InputDStream
import org.apache.spark.streaming.kafka010._
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.{SparkConf, broadcast}

/**
  * description:用户自定义风险库_分析
  **/
object DiyRelationArticleV5 {

  Logger.getLogger("org").setLevel(Level.ERROR)

  def main(args: Array[String]): Unit = {

    val conf = new SparkConf()
      .setAppName(this.getClass.getSimpleName)
      //      .setMaster("local[*]")
      .set("spark.streaming.kafka.maxRatePerPartition", args(0)) //每个分区每秒拉取500条数据
      .set("spark.streaming.stopGracefullyOnShutdown", "true")
      .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
      .registerKryoClasses(Array(classOf[NlpJsonBean]))

    val ssc = new StreamingContext(conf, Seconds(args(1).toInt))

    //获取到媒体数据并将数据进行广播
    val mediaMap: Map[String, String] = BroadcastTools.getMediaDic()
    val mediaDic: broadcast.Broadcast[Map[String, String]] = ssc.sparkContext.broadcast(mediaMap)

    val groupid = "g_diyrelation"

    //获取取kafka上的流数据
    val articleJson: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream[String, String](
      ssc,
      LocationStrategies.PreferConsistent, //将拉取到的数据，均匀分散到每台Executor节点上
      ConsumerStrategies.Subscribe[String, String](Array(ConfigHandler.topic), ConfigHandler.kafkaParams(groupid))
    )

    //kafka生产者
    val kafkaBro = ssc.sparkContext.broadcast(BroadcastKafkaProducer[String, String](ConfigHandler.kafkaProps))

    articleJson.foreachRDD(rdd => {

      //获取当前批次偏移量信息
      val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges

      //如果当前批次有数据，才继续以下操作
      if (!rdd.isEmpty()) {

        //将json数据转换成class
        val jsRDD: RDD[NlpJsonBean] = rdd.map(re => {
          JSON.parseObject(re.value(), classOf[NlpJsonBean])
        })

        jsRDD.foreachPartition(iter => {

          var conn: Connection = null
          var wordGroupPs: PreparedStatement = null
          var informationPs: PreparedStatement = null
          val gson = new Gson()

          try {

            conn = DriverManager.getConnection(ConfigHandler.url, ConfigHandler.user, ConfigHandler.passwd)
            iter.foreach(js => {

              //查询用户自定义词组表  看词组的启用和禁用状态，只选择启动run_flag=1，未删除del_flag=0的记录
              wordGroupPs = conn.prepareStatement(
                """
              select
              id,user_id,group_name
              from diy_keyword_group
              where run_flag = 1 and del_flag = 0
            """.stripMargin)

              val wordGroupSet: ResultSet = wordGroupPs.executeQuery()

              //取出json里面需要的值
              //媒体类型 1
              val mediaCls: String = js.media_cls
              //站点重要度 1
              val importantLeavel: String = js.important_leavel
              //站点网址 1
              val siteUrl: String = js.site_url
              //文章标题提取 1
              val articleTitle: String = js.article_title
              //文章内容 1
              val articleContent: String = js.article_content

              //遍历用户自定义风险词组表：diy_keyword_group
              while (wordGroupSet.next()) {

                val userId: String = wordGroupSet.getString("user_id")
                val groupId: String = wordGroupSet.getString("id")
                val groupName: String = wordGroupSet.getString("group_name")

                //词组信息表
                informationPs = conn.prepareStatement(
                  """
                  select
                  id,userid,filter_type,filter_key_site,region_words,subject_words,event_words,
                  exclusion_words,contain_urls,exclusion_urls,site_type
                  from diy_keyword_lines
                  where del_flag = 0 and group_id = ?
                """.stripMargin)

                informationPs.setString(1, groupId)

                val wordInfoSet = informationPs.executeQuery()

                //遍历筛选条件
                while (wordInfoSet.next()) {

                  val groupLineId = wordInfoSet.getString("id")

                  //媒体类型
                  val siteTpyeArr: Array[String] = wordInfoSet.getString("site_type").split(",")

                  //文章站点重要度
                  val filterKeySite = wordInfoSet.getInt("filter_key_site")

                  //判断站点网址
                  //val siteUrl = wordInfoSet.getString("site_url")

                  //禁止网站 可能是无
                  val exclusionUrls = wordInfoSet.getString("exclusion_urls")
                  //包含网站 可能是无
                  val containUrls = wordInfoSet.getString("contain_urls")
                  //地域关键词
                  val regionWords = wordInfoSet.getString("region_words")
                  //主题关键词
                  val subjectWords = wordInfoSet.getString("subject_words")
                  //时间关键词
                  val eventWords = wordInfoSet.getString("event_words")
                  //排除词 可能是无
                  val exclusionWords = wordInfoSet.getString("exclusion_words")

                  //判断匹配类型  仅标题？ 仅正文？ 标题或正文？
                  val filterType: Int = wordInfoSet.getInt("filter_type")

                  //筛选逻辑
                  //文章的（media_cls）媒体类型是否包含在用户自定义筛选条件的媒体类型site_type中
                  if (siteTpyeArr.contains(mediaCls)) {
                    //文章的（important_leavel）站点重要度是否是用户自定义的站点重要度filter_key_site
                    if (importantLeavel.equals(filterKeySite.toString)) {

                      //以下分四种情况
                      //1.包含网站列表不为空，禁止网站列表为空 且文章站点网址包含用户定义的包含网址
                      if (StringUtils.isNotEmpty(containUrls) && StringUtils.isEmpty(exclusionUrls) && DiyRelationArticleUtils.siteUrlContainsContainUrls(containUrls, siteUrl)) {

                        //内容变量
                        var content = ""
//                        val gson = new Gson()
                        //标题与内容
                        if (0 == filterType) {
                          content = articleTitle + articleContent
                          val value: String = DiyRelationArticleUtils.getValues(js, mediaDic, mediaCls, userId, groupId, groupName, groupLineId, regionWords, subjectWords, eventWords, exclusionWords, content)
                          println("标题与内容,包含网站列表不为空，禁止网站列表为空")
                          println(value)

                          if (StringUtils.isNotEmpty(value)) {

                            kafkaBro.value.send("ba_diy_risk_relation_article", UUID.randomUUID().toString.replaceAll("-", ""), value)
                            kafkaBro.value.send("ba_article", UUID.randomUUID().toString.replaceAll("-", ""), gson.toJson(BaArticle(js.uuid, js.article_title, js.article_content, js.nlp_emotion, "ba_article")))

                          }


                          //仅标题
                        } else if (1 == filterType) {
                          content = articleTitle
                          val value: String = DiyRelationArticleUtils.getValues(js, mediaDic, mediaCls, userId, groupId, groupName, groupLineId, regionWords, subjectWords, eventWords, exclusionWords, content)
                          println("标题,包含网站列表不为空，禁止网站列表为空")
                          println(value)

                          if (StringUtils.isNotEmpty(value)) {

                            kafkaBro.value.send("ba_diy_risk_relation_article", UUID.randomUUID().toString.replaceAll("-", ""), value)
                            kafkaBro.value.send("ba_article", UUID.randomUUID().toString.replaceAll("-", ""), gson.toJson(BaArticle(js.uuid, js.article_title, js.article_content, js.nlp_emotion, "ba_article")))

                          }
                          //仅正文
                        } else {
                          content = articleContent
                          val value: String = DiyRelationArticleUtils.getValues(js, mediaDic, mediaCls, userId, groupId, groupName, groupLineId, regionWords, subjectWords, eventWords, exclusionWords, content)
                          println("内容,包含网站列表不为空，禁止网站列表为空")
                          println(value)

                          if (StringUtils.isNotEmpty(value)) {

                            kafkaBro.value.send("ba_diy_risk_relation_article", UUID.randomUUID().toString.replaceAll("-", ""), value)
                            kafkaBro.value.send("ba_article", UUID.randomUUID().toString.replaceAll("-", ""), gson.toJson(BaArticle(js.uuid, js.article_title, js.article_content, js.nlp_emotion, "ba_article")))

                          }
                        }

                      }

                      //2.包含网站列表为空，禁止网站列表不为空 且文章站点网址不包含用户定义的禁止网址
                      if (StringUtils.isEmpty(containUrls) && StringUtils.isNotEmpty(exclusionUrls) && !DiyRelationArticleUtils.siteUrlContainsExclusionUrls(exclusionUrls, siteUrl)) {

                        //内容变量
                        var content = ""
//                        val gson = new Gson()
                        //标题与内容
                        if (0 == filterType) {
                          content = articleTitle + articleContent
                          val value: String = DiyRelationArticleUtils.getValues(js, mediaDic, mediaCls, userId, groupId, groupName, groupLineId, regionWords, subjectWords, eventWords, exclusionWords, content)
                          println("标题与内容,包含网站列表为空，禁止网站列表不为空")
                          println(value)

                          if (StringUtils.isNotEmpty(value)) {

                            kafkaBro.value.send("ba_diy_risk_relation_article", UUID.randomUUID().toString.replaceAll("-", ""), value)
                            kafkaBro.value.send("ba_article", UUID.randomUUID().toString.replaceAll("-", ""), gson.toJson(BaArticle(js.uuid, js.article_title, js.article_content, js.nlp_emotion, "ba_article")))

                          }
                          //仅标题
                        } else if (1 == filterType) {
                          content = articleTitle
                          val value: String = DiyRelationArticleUtils.getValues(js, mediaDic, mediaCls, userId, groupId, groupName, groupLineId, regionWords, subjectWords, eventWords, exclusionWords, content)
                          println("标题,包含网站列表为空，禁止网站列表不为空")
                          println(value)

                          if (StringUtils.isNotEmpty(value)) {

                            kafkaBro.value.send("ba_diy_risk_relation_article", UUID.randomUUID().toString.replaceAll("-", ""), value)
                            kafkaBro.value.send("ba_article", UUID.randomUUID().toString.replaceAll("-", ""), gson.toJson(BaArticle(js.uuid, js.article_title, js.article_content, js.nlp_emotion, "ba_article")))

                          }
                          //仅正文
                        } else {
                          content = articleContent
                          val value: String = DiyRelationArticleUtils.getValues(js, mediaDic, mediaCls, userId, groupId, groupName, groupLineId, regionWords, subjectWords, eventWords, exclusionWords, content)
                          println("内容,包含网站列表为空，禁止网站列表不为空")
                          println(value)

                          if (StringUtils.isNotEmpty(value)) {

                            kafkaBro.value.send("ba_diy_risk_relation_article", UUID.randomUUID().toString.replaceAll("-", ""), value)
                            kafkaBro.value.send("ba_article", UUID.randomUUID().toString.replaceAll("-", ""), gson.toJson(BaArticle(js.uuid, js.article_title, js.article_content, js.nlp_emotion, "ba_article")))

                          }
                        }

                      }

                      //3.都不为空 并符合条件
                      if (StringUtils.isNotEmpty(containUrls) && StringUtils.isNotEmpty(exclusionUrls) && DiyRelationArticleUtils.siteUrlContainsContainUrls(containUrls, siteUrl) && !DiyRelationArticleUtils.siteUrlContainsExclusionUrls(exclusionUrls, siteUrl)) {

                        //内容变量
                        var content = ""
//                        val gson = new Gson()
                        //标题与内容
                        if (0 == filterType) {
                          content = articleTitle + articleContent
                          val value: String = DiyRelationArticleUtils.getValues(js, mediaDic, mediaCls, userId, groupId, groupName, groupLineId, regionWords, subjectWords, eventWords, exclusionWords, content)
                          println("标题与内容,都不为空 并符合条件")
                          println(value)

                          if (StringUtils.isNotEmpty(value)) {

                            kafkaBro.value.send("ba_diy_risk_relation_article", UUID.randomUUID().toString.replaceAll("-", ""), value)
                            kafkaBro.value.send("ba_article", UUID.randomUUID().toString.replaceAll("-", ""), gson.toJson(BaArticle(js.uuid, js.article_title, js.article_content, js.nlp_emotion, "ba_article")))

                          }
                          //仅标题
                        } else if (1 == filterType) {
                          content = articleTitle
                          val value: String = DiyRelationArticleUtils.getValues(js, mediaDic, mediaCls, userId, groupId, groupName, groupLineId, regionWords, subjectWords, eventWords, exclusionWords, content)
                          println("标题,都不为空 并符合条件")
                          println(value)

                          if (StringUtils.isNotEmpty(value)) {

                            kafkaBro.value.send("ba_diy_risk_relation_article", UUID.randomUUID().toString.replaceAll("-", ""), value)
                            kafkaBro.value.send("ba_article", UUID.randomUUID().toString.replaceAll("-", ""), gson.toJson(BaArticle(js.uuid, js.article_title, js.article_content, js.nlp_emotion, "ba_article")))

                          }
                          //仅正文
                        } else {
                          content = articleContent
                          val value: String = DiyRelationArticleUtils.getValues(js, mediaDic, mediaCls, userId, groupId, groupName, groupLineId, regionWords, subjectWords, eventWords, exclusionWords, content)
                          println("内容,都不为空 并符合条件")
                          println(value)

                          if (StringUtils.isNotEmpty(value)) {

                            kafkaBro.value.send("ba_diy_risk_relation_article", UUID.randomUUID().toString.replaceAll("-", ""), value)
                            kafkaBro.value.send("ba_article", UUID.randomUUID().toString.replaceAll("-", ""), gson.toJson(BaArticle(js.uuid, js.article_title, js.article_content, js.nlp_emotion, "ba_article")))

                          }
                        }

                      }

                      //4.都为空
                      if (StringUtils.isEmpty(containUrls) && StringUtils.isEmpty(exclusionUrls)) {

                        //内容变量
                        var content = ""
//                        val gson = new Gson()
                        //标题与内容
                        if (0 == filterType) {
                          content = articleTitle + articleContent
                          val value: String = DiyRelationArticleUtils.getValues(js, mediaDic, mediaCls, userId, groupId, groupName, groupLineId, regionWords, subjectWords, eventWords, exclusionWords, content)
                          println("标题与内容,都为空")
                          println(value)

                          if (StringUtils.isNotEmpty(value)) {

                            kafkaBro.value.send("ba_diy_risk_relation_article", UUID.randomUUID().toString.replaceAll("-", ""), value)
                            kafkaBro.value.send("ba_article", UUID.randomUUID().toString.replaceAll("-", ""), gson.toJson(BaArticle(js.uuid, js.article_title, js.article_content, js.nlp_emotion, "ba_article")))

                          }
                          //仅标题
                        } else if (1 == filterType) {
                          content = articleTitle
                          val value: String = DiyRelationArticleUtils.getValues(js, mediaDic, mediaCls, userId, groupId, groupName, groupLineId, regionWords, subjectWords, eventWords, exclusionWords, content)
                          println("标题与内容,都为空")
                          println(value)

                          if (StringUtils.isNotEmpty(value)) {

                            kafkaBro.value.send("ba_diy_risk_relation_article", UUID.randomUUID().toString.replaceAll("-", ""), value)
                            kafkaBro.value.send("ba_article", UUID.randomUUID().toString.replaceAll("-", ""), gson.toJson(BaArticle(js.uuid, js.article_title, js.article_content, js.nlp_emotion, "ba_article")))

                          }
                          //仅正文
                        } else {
                          content = articleContent
                          val value: String = DiyRelationArticleUtils.getValues(js, mediaDic, mediaCls, userId, groupId, groupName, groupLineId, regionWords, subjectWords, eventWords, exclusionWords, content)
                          println("标题与内容,都为空")
                          println(value)

                          if (StringUtils.isNotEmpty(value)) {

                            kafkaBro.value.send("ba_diy_risk_relation_article", UUID.randomUUID().toString.replaceAll("-", ""), value)
                            kafkaBro.value.send("ba_article", UUID.randomUUID().toString.replaceAll("-", ""), gson.toJson(BaArticle(js.uuid, js.article_title, js.article_content, js.nlp_emotion, "ba_article")))

                          }
                        }

                      }

                    }
                  }
                }

              }

            })

          } catch {

            case e: Exception => e.printStackTrace()

          } finally {

            if (informationPs != null) informationPs.close()
            if (wordGroupPs != null) wordGroupPs.close()
            if (conn != null) conn.close()

          }

        })

      }

      //      提交偏移量信息
      articleJson.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges)

    })

    ssc.start()
    ssc.awaitTermination()

  }

}
